package com.workcell.workdata.im.component.message;

import com.farsunset.cim.group.SessionGroup;
import com.farsunset.cim.model.Message;
import com.workcell.workdata.im.component.event.BatchMessageEvent;
import com.workcell.workdata.im.component.event.ThirdMessagePushEvent;
import com.workcell.workdata.im.model.dto.BatchMessage;
import com.workcell.workdata.im.util.JSON;
import io.netty.channel.Channel;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.function.Predicate;


/**
 * 集群环境下，监听redis队列，广播消息到每个实例进行推送
 * 如果使用MQ的情况也，最好替换为MQ消息队列
 */
@Component
public class PushMessageListener implements MessageListener {

    @Resource
    private SessionGroup sessionGroup;

    @Resource
    private ApplicationContext applicationContext;

    @Override
    public void onMessage(org.springframework.data.redis.connection.Message redisMessage, byte[] bytes) {
        BatchMessage batchMessage = JSON.parse(redisMessage.getBody(), BatchMessage.class);
        this.handle(batchMessage);
    }

    @EventListener
    public void onMessage(BatchMessageEvent event) {
        this.handle(event.getSource());
    }

    private void handle(BatchMessage batchMessage){

        List<Message> messageList = batchMessage.getMessageList();

        Predicate<Channel> writePredicate = batchMessage.getPredicate();

        messageList.forEach(message -> sessionGroup.write(message.getReceiver(), message, writePredicate));

        messageList.removeIf(message -> sessionGroup.containsKey(message.getReceiver()));

        applicationContext.publishEvent(new ThirdMessagePushEvent(messageList, batchMessage.getTraceId()));
    }


}
